package com.flink.hbase;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * Flink HBase ETL 作业启动器
 * 从 Kerberos 认证的 HBase 读取数据，转换后写入 HDFS Parquet 文件
 */
public class JobLauncher {

    private static final Logger LOG = LoggerFactory.getLogger(JobLauncher.class);

    public static void main(String[] args) throws Exception {
        LOG.info("启动 Flink HBase ETL 作业");

        // 解析命令行参数
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        
        // 验证必需参数
        validateRequiredParameters(parameterTool);

        // 创建 Flink 执行环境
        StreamExecutionEnvironment env = createExecutionEnvironment(parameterTool);

        // 构建数据流处理管道
        buildDataPipeline(env, parameterTool);

        // 启动作业
        String jobName = parameterTool.get("job.name", "Flink-HBase-ETL");
        LOG.info("启动作业: {}", jobName);
        env.execute(jobName);
    }

    /**
     * 验证必需参数
     */
    private static void validateRequiredParameters(ParameterTool parameterTool) {
        String[] requiredParams = {
            "hbase.table.name",
            "output.path"
        };

        for (String param : requiredParams) {
            if (!parameterTool.has(param)) {
                throw new IllegalArgumentException("缺少必需参数: " + param);
            }
        }

        LOG.info("参数验证通过");
    }

    /**
     * 创建 Flink 执行环境
     */
    private static StreamExecutionEnvironment createExecutionEnvironment(ParameterTool parameterTool) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置全局参数
        env.getConfig().setGlobalJobParameters(parameterTool);

        // 设置并行度
        int parallelism = parameterTool.getInt("parallelism", 4);
        env.setParallelism(parallelism);
        LOG.info("设置并行度: {}", parallelism);

        // 启用检查点
        long checkpointInterval = parameterTool.getLong("checkpoint.interval", 60000);
        env.enableCheckpointing(checkpointInterval);
        LOG.info("启用检查点，间隔: {}ms", checkpointInterval);

        // 配置重启策略
        configureRestartStrategy(env, parameterTool);

        // 配置 Kerberos（如果启用）
        configureKerberos(env, parameterTool);

        return env;
    }

    /**
     * 配置重启策略
     */
    private static void configureRestartStrategy(StreamExecutionEnvironment env, ParameterTool parameterTool) {
        // 使用固定延迟重启策略
        int maxAttempts = parameterTool.getInt("restart.attempts", 3);
        long delayMs = parameterTool.getLong("restart.delay", 10000);
        
        env.setRestartStrategy(
            org.apache.flink.api.common.restartstrategy.RestartStrategies.fixedDelayRestart(
                maxAttempts, 
                org.apache.flink.api.common.time.Time.milliseconds(delayMs)
            )
        );
        
        LOG.info("配置重启策略: maxAttempts={}, delayMs={}", maxAttempts, delayMs);
    }

    /**
     * 配置 Kerberos
     */
    private static void configureKerberos(StreamExecutionEnvironment env, ParameterTool parameterTool) {
        boolean kerberosEnabled = parameterTool.getBoolean("security.kerberos.enabled", false);
        
        if (kerberosEnabled) {
            LOG.info("启用 Kerberos 认证");
            
            Configuration config = new Configuration();
            config.setString("security.kerberos.login.principal", 
                            parameterTool.get("security.kerberos.login.principal", ""));
            config.setString("security.kerberos.login.keytab", 
                            parameterTool.get("security.kerberos.login.keytab", ""));
            
            // 应用配置到环境
            env.configure(config);
            
            LOG.info("Kerberos 配置完成");
        } else {
            LOG.info("未启用 Kerberos 认证");
        }
    }

    /**
     * 构建数据处理管道
     */
    private static void buildDataPipeline(StreamExecutionEnvironment env, ParameterTool parameterTool) {
        // 1. 创建用户ID数据源
        DataStream<String> userIdStream = createUserIdSource(env, parameterTool);

        // 2. 异步查询 HBase 获取用户数据
        DataStream<User> userStream = createHBaseAsyncStream(userIdStream, parameterTool);

        // 3. 数据转换和清洗
        DataStream<User> cleanedUserStream = userStream
                .map(new UserMapper())
                .name("User Data Transformation")
                .filter(UserMapper::isValidUser)
                .name("Data Quality Filter");

        // 4. 写入 Parquet 文件
        writeToParquetFiles(cleanedUserStream, parameterTool);

        LOG.info("数据处理管道构建完成");
    }

    /**
     * 创建用户ID数据源
     */
    private static DataStream<String> createUserIdSource(StreamExecutionEnvironment env, ParameterTool parameterTool) {
        // 可以从多种数据源读取用户ID：Kafka、文件、数据库等
        String sourceType = parameterTool.get("source.type", "mock");
        
        switch (sourceType.toLowerCase()) {
            case "mock":
                return createMockUserIdSource(env, parameterTool);
            case "file":
                return createFileUserIdSource(env, parameterTool);
            case "kafka":
                // TODO: 实现 Kafka 数据源
                throw new UnsupportedOperationException("Kafka 数据源尚未实现");
            default:
                throw new IllegalArgumentException("不支持的数据源类型: " + sourceType);
        }
    }

    /**
     * 创建模拟用户ID数据源（用于测试）
     */
    private static DataStream<String> createMockUserIdSource(StreamExecutionEnvironment env, ParameterTool parameterTool) {
        int mockDataSize = parameterTool.getInt("mock.data.size", 1000);
        
        return env.addSource(new SourceFunction<String>() {
            private volatile boolean running = true;
            private int count = 0;

            @Override
            public void run(SourceContext<String> ctx) throws Exception {
                while (running && count < mockDataSize) {
                    String userId = "user_" + String.format("%06d", count + 1);
                    ctx.collect(userId);
                    count++;
                    
                    // 控制生成速率
                    Thread.sleep(10);
                }
                LOG.info("模拟数据源生成完成，总数: {}", count);
            }

            @Override
            public void cancel() {
                running = false;
            }
        }).name("Mock User ID Source");
    }

    /**
     * 创建文件用户ID数据源
     */
    private static DataStream<String> createFileUserIdSource(StreamExecutionEnvironment env, ParameterTool parameterTool) {
        String inputPath = parameterTool.getRequired("source.file.path");
        
        return env.readTextFile(inputPath)
                  .name("File User ID Source");
    }

    /**
     * 创建 HBase 异步查询流
     */
    private static DataStream<User> createHBaseAsyncStream(DataStream<String> userIdStream, ParameterTool parameterTool) {
        String tableName = parameterTool.getRequired("hbase.table.name");
        int asyncPoolSize = parameterTool.getInt("async.pool.size", 10);
        long timeoutMs = parameterTool.getLong("async.timeout.ms", 5000);
        
        HBaseAsyncSource hbaseSource = new HBaseAsyncSource(tableName, asyncPoolSize, timeoutMs);
        
        return AsyncDataStream.unorderedWait(
            userIdStream,
            hbaseSource,
            timeoutMs,
            TimeUnit.MILLISECONDS,
            asyncPoolSize
        ).name("HBase Async Query");
    }

    /**
     * 写入 Parquet 文件
     */
    private static void writeToParquetFiles(DataStream<User> userStream, ParameterTool parameterTool) {
        String outputPath = ParquetFileSink.ParquetSinkUtils.validateOutputPath(
            parameterTool.getRequired("output.path"));

        // 创建 Parquet Sink 配置
        ParquetFileSink.ParquetSinkConfig sinkConfig = ParquetFileSink.ParquetSinkUtils.createConfigFromParameters(
            parameterTool.get("sink.file.size.mb"),
            parameterTool.get("sink.rollover.interval.ms"),
            parameterTool.get("sink.inactivity.interval.ms"),
            parameterTool.get("sink.file.prefix"),
            parameterTool.get("sink.file.suffix"),
            parameterTool.get("sink.compression.codec")
        );

        // 创建 FileSink
        FileSink<User> fileSink = ParquetFileSink.createSink(outputPath, sinkConfig);

        // 设置 Sink 并行度
        int sinkParallelism = parameterTool.getInt("sink.parallelism", 
                                                  parameterTool.getInt("parallelism", 4));

        userStream.sinkTo(fileSink)
                  .name("Parquet File Sink")
                  .setParallelism(sinkParallelism);

        LOG.info("配置 Parquet 文件输出: path={}, config={}", outputPath, sinkConfig);
    }

    /**
     * 打印使用说明
     */
    public static void printUsage() {
        System.out.println("Flink HBase ETL 作业使用说明:");
        System.out.println();
        System.out.println("必需参数:");
        System.out.println("  --hbase.table.name <table>        HBase 表名");
        System.out.println("  --output.path <path>              输出 HDFS 路径");
        System.out.println();
        System.out.println("可选参数:");
        System.out.println("  --job.name <name>                 作业名称");
        System.out.println("  --parallelism <num>               并行度");
        System.out.println("  --source.type <type>              数据源类型 (mock|file|kafka)");
        System.out.println("  --checkpoint.interval <ms>        检查点间隔");
        System.out.println("  --async.pool.size <num>           异步线程池大小");
        System.out.println("  --async.timeout.ms <ms>           异步查询超时");
        System.out.println();
        System.out.println("Kerberos 参数:");
        System.out.println("  --security.kerberos.enabled <bool>      启用 Kerberos");
        System.out.println("  --security.kerberos.login.principal <p> 用户主体");
        System.out.println("  --security.kerberos.login.keytab <kt>    Keytab 文件路径");
        System.out.println();
        System.out.println("HBase 配置:");
        System.out.println("  --hbase.zookeeper.quorum <hosts>        ZooKeeper 地址");
        System.out.println("  --hbase.zookeeper.port <port>           ZooKeeper 端口");
        System.out.println();
        System.out.println("Sink 配置:");
        System.out.println("  --sink.file.size.mb <size>              文件大小限制(MB)");
        System.out.println("  --sink.compression.codec <codec>        压缩格式");
        System.out.println();
    }
} 